/*
 * Copyright 2022 The Open Islands Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.openislands.oi.worker.scheduler;

import org.openislands.oi.constant.Dict;
import org.openislands.oi.constant.LogTraceType;
import org.openislands.oi.constant.LogType;
import org.openislands.oi.constant.TaskState;
import org.openislands.oi.util.CustomerLogger;
import org.openislands.oi.util.CustomerLoggerFactory;
import org.openislands.oi.worker.manager.OfNodeTaskManager;
import org.openislands.oi.dao.entity.OfNodeTaskDO;
import org.openislands.oi.pojo.dto.OfNodeTaskDTO;
import org.openislands.oi.scheduling.StateEngine;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

@SuppressWarnings("unused")
@Service
public class OperatorRuntimeWatcher {
    private static final CustomerLogger monitorLogger = CustomerLoggerFactory.getLogger(LogTraceType.JOB, LogType.MONITOR);

    @Resource
    private OfNodeTaskManager ofNodeTaskManager;

    @Resource
    private StateEngine stateEngine;

    @Resource
    private OperatorRuntimeManager operatorRuntimeManager;

    @Value(Dict.OPERATOR_RUNTIME_HEARTBEAT_TIMEOUT)
    private int operatorRuntimeHeartbeatTimeout;
    @Value(Dict.OPERATOR_RUNTIME_LAUNCH_TIMEOUT)
    private int operatorRuntimeLaunchTimeout;

    public void checkOperatorRuntime() {
        monitorLogger.info("OperatorRuntimeWatcher.checkOperatorRuntime-------------------start");
        // query unfinished and heartbeat timeout node task
        Date currentDate = new Date();
        Calendar cl = Calendar.getInstance();
        cl.setTime(currentDate);
        cl.add(Calendar.SECOND, -operatorRuntimeLaunchTimeout);
        final Set<TaskState> unfinishedTaskStates = stateEngine.getUnfinishedTaskStates();
        List<OfNodeTaskDTO> ofNodeTaskDTOList = ofNodeTaskManager.queryConditionOfNodeTask(
                lambdaWrapper -> lambdaWrapper.in(OfNodeTaskDO::getState, unfinishedTaskStates).
                        isNotNull(OfNodeTaskDO::getLatestHeartbeat).
                        lt(OfNodeTaskDO::getLatestHeartbeat, cl.getTime())
        );
        for (OfNodeTaskDTO ofNodeTaskDTO : ofNodeTaskDTOList) {
            try {
                if (operatorRuntimeManager.aliveRuntime(ofNodeTaskDTO)){
                    long t = currentDate.getTime() - ofNodeTaskDTO.getLatestHeartbeat().getTime();
                    if(t > operatorRuntimeHeartbeatTimeout){
                        monitorLogger.error(ofNodeTaskDTO.getJobId(), "the latest heartbeat of task {} is {} and timeout according to the timeout limit: {}", ofNodeTaskDTO.getSessionId(), ofNodeTaskDTO.getLatestHeartbeat(), operatorRuntimeHeartbeatTimeout);
                    }else{
                        continue;
                    }
                }else {
                    monitorLogger.error(ofNodeTaskDTO.getJobId(), "task {} runtime is not alive", ofNodeTaskDTO.getSessionId());
                }
                try {
                    operatorRuntimeManager.terminateRuntime(ofNodeTaskDTO);
                } catch (Exception ignored) {
                }
                // runtime is alive
                if (!operatorRuntimeManager.aliveRuntime(ofNodeTaskDTO)) {
                    OfNodeTaskDTO update = new OfNodeTaskDTO();
                    update.setState(TaskState.RUNTIME_ABORTED);
                    Boolean aBoolean = operatorRuntimeManager.updateRuntimeState(
                            ofNodeTaskDTO.getJobId(), ofNodeTaskDTO.getTaskName(), ofNodeTaskDTO.getTaskVersion(),
                            ofNodeTaskDTO.getRole(), ofNodeTaskDTO.getNodeId(), update
                    );
                    if (!aBoolean) {
                        monitorLogger.warn(ofNodeTaskDTO.getJobId(), "task {} update runtime state failed", ofNodeTaskDTO.getSessionId());
                    }
                }else {
                    monitorLogger.warn(ofNodeTaskDTO.getJobId(), "task {} runtime remains alive after being terminated, waiting for the next round", ofNodeTaskDTO.getSessionId());
                }
            } catch (Exception e) {
                monitorLogger.error(ofNodeTaskDTO.getJobId(), "check task {} runtime failed", ofNodeTaskDTO.getSessionId());
            }
        }
        monitorLogger.info("OperatorRuntimeWatcher.checkOperatorRuntime-------------------end");
    }
}
